package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用状态State编程，对is_new字段值进行校验和纠正
 */
public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String,String,String> {

    //step1-声明状态
    private transient ValueState<String> firstDateState =null;


    @Override
    public void open(Configuration parameters) throws Exception {
        //step2-初始化状态
        firstDateState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("first_date", String.class)
        );

    }

    @Override
    public void processElement(String s, KeyedProcessFunction<String, String, String>.Context context, Collector<String> collector) throws Exception {
        //step3-firstDate
        String firstDate = firstDateState.value();

        //解析字段值
        JSONObject jsonObject = JSON.parseObject(s);
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        Long tsValue = jsonObject.getLong("ts");

        //TODO isNew=1 --->>>firstDate=null firstDate=有值
        if("1".equals(isNewValue)){
            if(null==firstDate){
                //step4-更新状态值
                //此时is_new正确，状态值更新为第一次访问日期
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue,DateTimeUtil.DATE_FORMAT));
            }else {
                //状态值（首次访问日期）等于数据日期，不做处理
                if(!firstDate.equals(DateTimeUtil.convertLongToString(tsValue,DateTimeUtil.DATE_FORMAT))){
                    //此时is_new属于异常，需要纠正，设置为0
                    jsonObject.getJSONObject("common").put("is_new","0");

                }
            }
        }

        //is_new=0 ---->>> firstDate=null firstDate=有值
        else {
            //当用户最早登录访问，当时实时应用后来处理，用户第一次访问时，firstDate为null
            if(null==firstDate){
                //取前一天日期
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue-24L * 60 * 60 * 1000,DateTimeUtil.DATE_FORMAT));
            }
        }

        //输出结果
        collector.collect(jsonObject.toJSONString());

    }
}

